﻿using System;
using System.Collections.Generic;
using System.Threading;
using ZooKeeperNet;

namespace Cluster
{
    public abstract partial class ClusterManager
    {
        private static readonly TimeSpan DEFAULT_TIMEOUT = TimeSpan.FromSeconds(5);

        private ZooKeeper zk;

        /// <summary>
        /// Starts a new Zoo Keeper connection
        /// </summary>
        /// <param name="zkConnectionString"></param>
        /// <param name="zkTimeout"></param>
        private void StartZk(string zkConnectionString, TimeSpan? zkTimeout)
        {
            var timeout = zkTimeout.HasValue ? zkTimeout.Value : DEFAULT_TIMEOUT;
            zk = new ZooKeeper(zkConnectionString, timeout, new RootWatcher(connectionEvent));

            try
            {
                connectionEvent.WaitOne(timeout);
            }
            catch (TimeoutException)
            {
                Error("Connection event timeout");
                throw new ClusterException("Failed to connect to ZooKeeper server");
            }
        }

        /// <summary>
        /// Kills the Zoo Keeper connection
        /// </summary>
        private void StopZk()
        {
            if (zk != null)
            {
                zk.Dispose();
                zk = null;
            }
        }

        /// <summary>
        /// Create Zoo Keeper path, if not already exists
        /// </summary>
        /// <param name="zpath"></param>
        /// <param name="persistent"></param>
        /// <param name="sequential"></param>
        private void CreateZkPath(string zpath, bool persistent = false, bool sequential = false)
        {
            zpath = zpath.TrimStart('/');

            var current = string.Empty;
            
            foreach (var pathElement in zpath.Split('/'))
            {
                current += ("/" + pathElement);
                
                var clousure = current;

                if (RetryOperation(() => zk.Exists(clousure, false)) == null)
                {
                    try
                    {
                        RetryOperation(() => zk.Create(clousure, null, Ids.OPEN_ACL_UNSAFE, GetCreationMode(persistent, sequential)));

                        Debug("Created ZK path " + current);
                    }
                    catch (KeeperException e)
                    {
                        if (e.ErrorCode != KeeperException.Code.NODEEXISTS) // might happen if not synchronized
                        {
                            ErrorEx("Failed to create ZK path: " + current, e);

                            throw new ClusterException("Failed to create ZK path", e);
                        }
                    }
                }
            }
        }

        /// <summary>
        /// Crates a new ZooKeeper node
        /// </summary>
        /// <param name="zpath"></param>
        /// <param name="persistent"></param>
        /// <param name="sequential"></param>
        /// <param name="data"></param>
        /// <returns></returns>
        private string CreateZkNode(string zpath, bool persistent = false, bool sequential = false, byte[] data = null)
        {
            try
            {
                var jpath = RetryOperation(() => zk.Create(zpath, data, Ids.OPEN_ACL_UNSAFE, GetCreationMode(persistent, sequential)));
                
                Debug("Added ZK node " + jpath);

                return jpath;
            }
            catch (KeeperException e)
            {
                ErrorEx("Failed to add ZK node", e);

                throw new ClusterException("Failed to add node", e);
            }
        }

        /// <summary>
        /// Removed Zoo Keeper path or node
        /// </summary>
        /// <param name="zpath"></param>
        private void RemoveZkPath(string zpath)
        {
            try
            {
                RetryOperation(() => {
                    zk.Delete(zpath, -1);
                    return 0; // dummy for generic func
                });
            }
            catch (KeeperException e)
            {
                ErrorEx("Failed to remove ZK path", e);

                throw new ClusterException("Failed to remove path", e);
            }
        }

        /// <summary>
        /// Updates the Zoo Keeper node data
        /// </summary>
        /// <param name="zpath"></param>
        /// <param name="data"></param>
        /// <param name="version"></param>
        private void SetZkNodeData(string zpath, byte[] data, int version = 0)
        {
            try
            {
                RetryOperation(() => zk.SetData(zpath, data, version));
            }
            catch (KeeperException e)
            {
                throw new ClusterException("Failed to set node data", e);
            }
        }

        /// <summary>
        /// Retrieves the Zoo Keeper node data
        /// </summary>
        /// <param name="zpath"></param>
        /// <returns></returns>
        private byte[] GetZkNodeData(string zpath)
        {
            try
            {
                return RetryOperation(() => zk.GetData(zpath, false, null));
            }
            catch (KeeperException e)
            {
                throw new ClusterException("Failed to get node data", e);
            }
        }

        /// <summary>
        /// Sets a watcher on the Zoo Keeper path
        /// </summary>
        /// <param name="zpath"></param>
        /// <param name="watcher"></param>
        private void WatchZkPath(string zpath, IWatcher watcher)
        {
            try
            {
                RetryOperation(() => zk.GetChildren(zpath, watcher));
            }
            catch (KeeperException e)
            {
                throw new ClusterException("Failed to set watch to path", e);
            }
        }

        /// <summary>
        /// Retrieves the child nodes of the specified path
        /// </summary>
        /// <param name="zpath"></param>
        /// <returns></returns>
        private List<string> GetZkChildren(string zpath)
        {
            try
            {
                return new List<string>(RetryOperation(() => zk.GetChildren(zpath, false)));
            }
            catch(KeeperException e)
            {
                throw new ClusterException("Failed to get path children", e);
            }
        }

        /// <summary>
        /// Performs the specified operation several until succeeded or limit is reached
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="operation"></param>
        /// <param name="retryCount"></param>
        /// <param name="retryDelaySec"></param>
        /// <returns></returns>
        protected T RetryOperation<T>(Func<T> operation, int retryCount = 5, int retryDelaySec = 1)
        {
            var exception = KeeperException.Create(KeeperException.Code.OPERATIONTIMEOUT);

            for (int i = 0; i < retryCount; ++i)
            {
                try
                {
                    return operation();
                }
                catch (KeeperException.SessionExpiredException)
                {
                    Warn("Session expired");

                    throw;
                }
                catch (KeeperException.ConnectionLossException e)
                {
                    exception = e;
                    
                    Debug("Attempt {0} failed with connection loss so attempting to reconnect", i+1);

                    Thread.Sleep(TimeSpan.FromSeconds(retryDelaySec));
                
                }
                catch (TimeoutException)
                {
                    Debug("Attempt {0} failed with connection loss so attempting to reconnect", i+1);

                    Thread.Sleep(TimeSpan.FromSeconds(retryDelaySec));
                }
            }

            throw exception;
        }
        
        /// <summary>
        /// Gets the Zoo Keeper creation mode by attributes
        /// </summary>
        /// <param name="persistent"></param>
        /// <param name="sequential"></param>
        /// <returns></returns>
        private CreateMode GetCreationMode(bool persistent, bool sequential)
        {
            return persistent
                ? (sequential ? CreateMode.PersistentSequential : CreateMode.Persistent)
                : (sequential ? CreateMode.EphemeralSequential : CreateMode.Ephemeral);
        }
    }
}
